package StructuredStreaming;
/**
 * scalable 可比较的  可攀登的
 * computation 计算
 * optimized 优化
 * reasoning 推理  推论
 * aggregation  聚合
 * semantics 语义  语义论   Falut Tolerance Semantics
 * delivering 提供
 * reliably  可靠地 确实地
 * idempotent 幂等
 *  recurse 递归
 * subdirectories 分录
 * straightforward 简单
 * intermediate  中间区域  中间地带  中间区
 * accumulate
 * specifying 指定  具体说明
 * watermarking 水印
 * threshold 阈值  门槛
 * dictated 口述的  口述
 * fine-grained 细粒的，深入的
 * arbitrary  任意的 武断的  专制的
 *
 */

/**
 *容错语义：  通过提供可重复启动的源，以及通过checkpoint 以及watermarker记录 stream source 的 offset以及设计消费的幂等性
 *
 */

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;


import java.util.Arrays;





/**
 * @program: MySpark
 * @description
 * @author: tkk fendoukaoziji
 * @create: 2019-04-20 11:49
 **/
public class ProgrammingModel {
    public static void main(String[] args) throws StreamingQueryException {
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN); //设置日志级别warn

        SparkSession spark = SparkSession
                .builder()
                .appName("JavaStructuredNetworkWordCount").master("local[2]")
                .getOrCreate();
        //create dataframe input lines from connection to node3:9999
        Dataset<Row> lines = spark
                .readStream()
                .format("socket")
                .option("host", "node3")
                .option("port", 9999)
                .load();
        Dataset<String> words = lines.as(Encoders.STRING()).flatMap((FlatMapFunction<String, String>) x -> Arrays.asList(x.split(" ")).iterator(), Encoders.STRING());
        Dataset<Row> wordCounts = words.groupBy("value").count();
        StreamingQuery query = wordCounts.writeStream()
                //.outputMode("update")   每次 只有输入流中变化的数据才会写出到流  如果不是aggregate操作 会和append一样
                //.outputMode("complete")// 适用于聚合统计 且每次更新时  所有流数据所有行写入sink
                .outputMode("append") //适用于 上次处理间隔以后新增加的  会写入结果集
                .format("console") //输出到控制台
                .start();
        query.awaitTermination();


    }
}
